package com.shujia.spark.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo4foreachPartitions {
  def main(args: Array[String]): Unit = {

    val startTIme: Long = System.currentTimeMillis()

    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext

    val studentsRDD: RDD[String] = sc.textFile("data/students.txt")


    /**
     * 将rdd的数据保存到mysql中
     *
     */


    /**
     * foreach: 每一条数据都需要创建一个网络链接
     * 不能将网络链接放在算子外（网络链接不能在网络中传输）
     *
     */
    /* studentsRDD.foreach(stu => {
       val split: Array[String] = stu.split(",")
       //1、加载启动
       Class.forName("com.mysql.jdbc.Driver")
       val start: Long = System.currentTimeMillis()
       //2、创建链接
       val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8", "root", "123456")
       val end: Long = System.currentTimeMillis()
       println(s"创建数据库的链接用了：${end - start}")
       //3、编写插入数据的sql
       val stat: PreparedStatement = con.prepareStatement("insert into students(id,name,age,gender,clazz) values(?,?,?,?,?)")
       //4、设置列值
       stat.setLong(1, split(0).toLong)
       stat.setString(2, split(1))
       stat.setLong(3, split(2).toLong)
       stat.setString(4, split(3))
       stat.setString(5, split(4))

       //5、执行插入
       stat.execute()

       //6、关闭链接
       stat.close()
       con.close()
     })*/

    /**
     * foreachPartition: 一次遍历一个分区的数据
     * 如果需要将rdd的数据保存到外部数据库中，比如mysql,hbase,redis, 需要使用foreachPartition
     */
    studentsRDD.foreachPartition(iter => {

      //1、加载启动
      Class.forName("com.mysql.jdbc.Driver")
      val start: Long = System.currentTimeMillis()
      //2、创建链接
      val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8", "root", "123456")
      val end: Long = System.currentTimeMillis()
      println(s"创建数据库的链接用了：${end - start}")
      //3、编写插入数据的sql
      val stat: PreparedStatement = con.prepareStatement("insert into students(id,name,age,gender,clazz) values(?,?,?,?,?)")

      iter.foreach(stu => {
        val split: Array[String] = stu.split(",")
        //4、设置列值
        stat.setLong(1, split(0).toLong)
        stat.setString(2, split(1))
        stat.setLong(3, split(2).toLong)
        stat.setString(4, split(3))
        stat.setString(5, split(4))

        //5、执行插入
        stat.execute()
      })
      //6、关闭链接
      stat.close()
      con.close()
    })


    val endTIme: Long = System.currentTimeMillis()

    println(s"共用了：${endTIme - startTIme}")
  }

}
